package drds.binlog.parse.mysql;

import com.lmax.disruptor.*;
import drds.binlog.binlog_event.LogByteBlock;
import drds.binlog.binlog_event.LogEvent;
import drds.binlog.common.AbstractLifeCycle;
import drds.binlog.common.Authors;
import drds.binlog.database_driver.packets.IGtidSet;
import drds.binlog.parse.EntryListTransaction;
import drds.binlog.parse.IConnection;
import drds.binlog.parse.IMultiStageCoprocessor;
import drds.binlog.parse.exception.ParseException;
import drds.binlog.parse.mysql.dbsync.BinlogLogEventConverter;
import drds.common.Author;
import lombok.Getter;
import lombok.Setter;

import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.atomic.AtomicLong;
import java.util.concurrent.locks.LockSupport;

/**
 * 针对解析器提供一个多阶段协同的处理
 * <p>
 * <pre>
 * 1. 网络接收 (单线程)
 * 2. 事件基本解析 (单线程，事件类型、DDL解析构造TableMeta、维护位点信息)
 * 3. 事件深度解析 (多线程, DML事件数据的完整解析)
 * 4. 投递到store (单线程)
 * </pre>
 */
@Author(name = Authors.LI_YANG)
public class MultiStageCoprocessor extends AbstractLifeCycle implements IMultiStageCoprocessor
{

    private static final int maxFullTimes = 10;
    @Setter
    @Getter
    public BinlogLogEventConverter binlogLogEventConverter;
    @Setter
    @Getter
    public EntryListTransaction entryListTransaction;
    @Setter
    @Getter
    public IConnection connection;
    @Setter
    @Getter
    public volatile ParseException parseException;
    @Setter
    @Getter
    public IGtidSet gtidSet;
    @Setter
    @Getter
    private int parserThreadCount;
    @Setter
    @Getter
    private int ringBufferSize;
    @Setter
    @Getter
    private RingBuffer<MessageEvent> messageEventRingBuffer;
    @Setter
    @Getter
    private ExecutorService parserExecutor;
    @Setter
    @Getter
    private ExecutorService stageExecutor;
    @Setter
    @Getter
    private String destination;
    @Setter
    @Getter
    private AtomicLong eventsPublishBlockingTime;

    public MultiStageCoprocessor(int ringBufferSize, int parserThreadCount, BinlogLogEventConverter binlogLogEventConverter,
                                 EntryListTransaction entryListTransaction, String destination)
    {
        this.ringBufferSize = ringBufferSize;
        this.parserThreadCount = parserThreadCount;
        this.binlogLogEventConverter = binlogLogEventConverter;
        this.entryListTransaction = entryListTransaction;
        this.destination = destination;
    }

    @Override
    public void start()
    {
        super.start();
        this.parseException = null;
        this.messageEventRingBuffer = RingBuffer.createSingleProducer(new MessageEventFactory(this),
                ringBufferSize,
                new BlockingWaitStrategy());
        int tc = parserThreadCount > 0 ? parserThreadCount : 1;
        this.parserExecutor = Executors.newFixedThreadPool(tc);

        this.stageExecutor = Executors.newFixedThreadPool(2);
        SequenceBarrier sequenceBarrier = messageEventRingBuffer.newBarrier();
        ExceptionHandler exceptionHandler = new SimpleFatalExceptionHandler();
        // stage 2
        BatchEventProcessor<MessageEvent> simpleParserStage = new BatchEventProcessor<MessageEvent>(messageEventRingBuffer,
                sequenceBarrier,
                new SimpleParserStage(this));
        simpleParserStage.setExceptionHandler(exceptionHandler);
        messageEventRingBuffer.addGatingSequences(simpleParserStage.getSequence());

        // stage 3
        SequenceBarrier dmlParserSequenceBarrier = messageEventRingBuffer.newBarrier(simpleParserStage.getSequence());
        WorkHandler<MessageEvent>[] workHandlers = new DmlParserStage[parserThreadCount];
        for (int i = 0; i < parserThreadCount; i++)
        {
            workHandlers[i] = new DmlParserStage(this);
        }
        WorkerPool<MessageEvent> workerPool = new WorkerPool<MessageEvent>(messageEventRingBuffer,
                dmlParserSequenceBarrier,
                exceptionHandler,
                workHandlers);
        Sequence[] sequence = workerPool.getWorkerSequences();
        messageEventRingBuffer.addGatingSequences(sequence);

        // stage 4
        SequenceBarrier sinkSequenceBarrier = messageEventRingBuffer.newBarrier(sequence);
        BatchEventProcessor<MessageEvent> sinkStoreStage = new BatchEventProcessor<MessageEvent>(messageEventRingBuffer,
                sinkSequenceBarrier,
                new SinkStoreStage(this));
        sinkStoreStage.setExceptionHandler(exceptionHandler);
        messageEventRingBuffer.addGatingSequences(sinkStoreStage.getSequence());

        // start work
        stageExecutor.submit(simpleParserStage);
        stageExecutor.submit(sinkStoreStage);
        workerPool.start(parserExecutor);
    }

    @Override
    public void stop()
    {
        try
        {
            parserExecutor.shutdownNow();
        } catch (Throwable e)
        {
            // ignore
        }

        try
        {
            stageExecutor.shutdownNow();
        } catch (Throwable e)
        {
            // ignore
        }
        super.stop();
    }

    public boolean publish(LogByteBlock logByteBlock)
    {
        return this.publish(logByteBlock, null);
    }

    /**
     * 网络数据投递
     */
    public boolean publish(LogEvent logEvent)
    {
        return this.publish(null, logEvent);
    }

    private boolean publish(LogByteBlock logByteBlock, LogEvent logEvent)
    {
        if (!isRunning())
        {
            if (parseException != null)
            {
                throw parseException;
            }
            return false;
        }

        /**
         * 由于改为processor仅终止自身stage而不是stop，那么需要由incident标识coprocessor是否正常工作。
         * 让dump线程能够及时感知
         */
        if (parseException != null)
        {
            throw parseException;
        }
        boolean interupted = false;
        long blockingStart = 0L;
        int fullTimes = 0;
        do
        {
            try
            {
                long next = messageEventRingBuffer.tryNext();
                MessageEvent messageEvent = messageEventRingBuffer.get(next);
                if (logByteBlock != null)
                {
                    messageEvent.setLogByteBlock(logByteBlock);
                } else
                {
                    messageEvent.setLogEvent(logEvent);
                }
                messageEventRingBuffer.publish(next);
                if (fullTimes > 0)
                {
                    eventsPublishBlockingTime.addAndGet(System.nanoTime() - blockingStart);
                }
                break;
            } catch (InsufficientCapacityException e)
            {
                if (fullTimes == 0)
                {
                    blockingStart = System.nanoTime();
                }
                // park
                // LockSupport.parkNanos(1L);
                applyWait(++fullTimes);
                interupted = Thread.interrupted();
                if (fullTimes % 1000 == 0)
                {
                    long nextStart = System.nanoTime();
                    eventsPublishBlockingTime.addAndGet(nextStart - blockingStart);
                    blockingStart = nextStart;
                }
            }
        } while (!interupted && isRunning());
        return isRunning();
    }

    // 处理无数据的情况，避免空循环挂死
    private void applyWait(int fullTimes)
    {
        int newFullTimes = fullTimes > maxFullTimes ? maxFullTimes : fullTimes;
        if (fullTimes <= 3)
        { // 3次以内
            Thread.yield();
        } else
        { // 超过3次，最多只sleep 1ms
            LockSupport.parkNanos(100 * 1000L * newFullTimes);
        }

    }

    @Override
    public void reset()
    {
        if (isRunning())
        {
            stop();
        }

        start();
    }


}
